package com.binance.assetservice.web.mq;

import com.binance.assetservice.web.service.IProductService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.javasimon.aop.Monitored;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Slf4j
@Component
@Monitored
public class RefreshProductListener {
    @Autowired
    private IProductService productService;

    /**
     * Discription: 处理MQ消息的监听方法.
     * 该监听方法没有指定具体的queue-name是由系统自动分配名字.'rabbitListenerContainerFactory'spring自动配置
     *
     * @param msg 自定义Obj2JsonMessageConverter返回的消息
     * @param channel the channel 接收对应的channel对象用于手动ack
     * @param deliveryTag 消息标记
     */
    @RabbitListener(
            bindings = @QueueBinding(value = @Queue(durable = "false", autoDelete = "true"),
                    exchange = @Exchange(value = "product_refresh", type = ExchangeTypes.FANOUT)),
            containerFactory = "rabbitListenerContainerFactory")
    public void receiveToRefreshProduct(String msg, @Header(AmqpHeaders.CHANNEL) Channel channel,
            @Header(AmqpHeaders.DELIVERY_TAG) Long deliveryTag) {
        try {
            if (StringUtils.isBlank(msg)) {
                log.warn("------>receive() The mq msg of received is null !");
                return;
            }
            log.info("------>receive() Receive message :" + msg);
            if (StringUtils.contains(msg, "#")) {
                String[] array = StringUtils.split(msg, "#");
                if (array.length == 2) {
                    productService.refreshProduct(array[1], false);
                } else {
                    productService.refreshProductAll(false);
                }
            } else {
                productService.refreshProductAll(false);
            }
        } catch (Exception e) {
            log.error("------>receive() Message parsing exception :", e);
        }
        try {
            // 手动ack确认
            channel.basicAck(deliveryTag, false);
        } catch (IOException e) {
            log.error("------>receive() Rabbit channel ack error :", e);
        }
    }

    @RabbitListener(
            bindings = @QueueBinding(value = @Queue(durable = "false", autoDelete = "true"),
                    exchange = @Exchange(value = "user_ProductTradeFee_refresh", type = ExchangeTypes.FANOUT)),
            containerFactory = "rabbitListenerContainerFactory")
    public void receiveToRefreshUserProductTradeFee(String msg, @Header(AmqpHeaders.CHANNEL) Channel channel,
            @Header(AmqpHeaders.DELIVERY_TAG) Long deliveryTag) {
        try {
            if (StringUtils.isBlank(msg)) {
                log.warn("------>receive() The mq msg of received is null !");
                return;
            }
            log.info("------>receive() Receive message :" + msg);
            if (StringUtils.contains(msg, "#")) {
                String[] array = StringUtils.split(msg, "#");
                this.productService.refreshUserTradeFee(Long.parseLong(array[0]), array[1], false);
            } else {
                this.productService.invalidateAllUserTradeFee(false);
            }
        } catch (Exception e) {
            log.error("------>receive() Message parsing exception :", e);
        }
        try {
            // 手动ack确认
            channel.basicAck(deliveryTag, false);
        } catch (IOException e) {
            log.error("------>receive() Rabbit channel ack error :", e);
        }
    }

}
